package com.offcn.bigdata.spark.streaming.p2

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * transform的transformation操作
  *  streaming中的join操作，但是该join操作仅限于dstream.join(dstream)
  *  如果要进行dstream和rdd的join操作，是没有这样的api，
  *  通过foreachRDD可以完成join关联，但是foreachRDD没有返回值，为了满足于此的操作。
  *  官方就提供了这样一个api：transform，可以把它理解为有返回值的foreachRDD
  *
  * 黑名单过滤的案例：
  *     数据：
  *         27.19.74.143##2016-05-30 17:38:20##GET /static/image/common/faq.gif HTTP/1.1##200##1127
  *     黑名单:
  *         27.19.74.143
  *         110.52.250.126
  *         数据结构：ip->是否激活/有效
  *     需求：
  *         过滤掉黑名单中相应的数据
  *
  */
object _03TransformOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName("_02StreamingWindowOps")
            .set("spark.streaming.kafka.maxRatePerPartition", "10")
        val batchInterval = 2
        val ssc = new StreamingContext(conf, Seconds(batchInterval))

        //黑名单
        val blacklist:RDD[(String, Boolean)] = ssc.sparkContext.parallelize(List(
            ("27.19.74.143", true),
            ("110.52.250.126", true),
            ("8.35.201.163", false)
        ))
        //加载流量数据

        val lines = ssc.socketTextStream("bigdata01", 9999)

        ssc.start()
        ssc.awaitTermination()
    }

    def filterByForeachRDD(lines:DStream[String], blacklist:RDD[(String, Boolean)]): Unit = {
        lines.foreachRDD(rdd => {
            if(!rdd.isEmpty()) {
                val ip2Info:RDD[(String, String)] = rdd.map(line => {
                    val index = line.indexOf("##")
                    val ip = line.substring(0, index)
                    val info = line.substring(index + 2)
                    (ip, info)
                })
                //关联
                val joined:RDD[(String, (String, Option[Boolean]))] = ip2Info.leftOuterJoin(blacklist)
                //过滤黑名单
                joined.filter{case (ip, (info, option)) => {
                    !option.isDefined || !option.get
                }}.map{case (ip, (info, option)) => (ip, info)}
                    .foreach(println)
            }
        })
    }
    def filterByTransform(lines:DStream[String], blacklist:RDD[(String, Boolean)]): Unit = {
        val filteredDStream:DStream[(String, String)] = lines.transform(rdd => {

            val ip2Info:RDD[(String, String)] = rdd.map(line => {
                val index = line.indexOf("##")
                val ip = line.substring(0, index)
                val info = line.substring(index + 2)
                (ip, info)
            })
            //关联
            val joined:RDD[(String, (String, Option[Boolean]))] = ip2Info.leftOuterJoin(blacklist)
            //过滤黑名单
            joined.filter{case (ip, (info, option)) => {
                !option.isDefined || !option.get
            }}.map{case (ip, (info, option)) => (ip, info)}
        })

        filteredDStream.print()
    }
}
